package com.wugui.datax.rpc.remoting.invoker;

import cn.hutool.core.collection.CollUtil;
import com.wugui.datax.rpc.registry.ServiceRegistry;
import com.wugui.datax.rpc.registry.impl.LocalServiceRegistry;
import com.wugui.datax.rpc.remoting.net.params.BaseCallback;
import com.wugui.datax.rpc.remoting.net.params.XxlRpcFutureResponse;
import com.wugui.datax.rpc.remoting.net.params.XxlRpcResponse;
import com.wugui.datax.rpc.util.XxlRpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * xxl-rpc invoker factory, init service-registry
 *
 * @author xuxueli 2018-10-19
 */
public class XxlRpcInvokerFactory {
  private static final Logger logger = LoggerFactory.getLogger(XxlRpcInvokerFactory.class);

  // ---------------------- default instance ----------------------

  private static final XxlRpcInvokerFactory instance =
      new XxlRpcInvokerFactory(LocalServiceRegistry.class, null);
  private final List<BaseCallback> stopCallbackList = new ArrayList<BaseCallback>();

  // ---------------------- config ----------------------
  private final ConcurrentMap<String, XxlRpcFutureResponse> futureResponsePool =
      new ConcurrentHashMap<String, XxlRpcFutureResponse>();
  private Class<? extends ServiceRegistry> serviceRegistryClass; // class.forname
  private Map<String, String> serviceRegistryParam;
  private ServiceRegistry serviceRegistry;

  // ---------------------- start / stop ----------------------
  private volatile ThreadPoolExecutor responseCallbackThreadPool = null;

  public XxlRpcInvokerFactory() {}

  // ---------------------- service registry ----------------------

  public XxlRpcInvokerFactory(
      Class<? extends ServiceRegistry> serviceRegistryClass,
      Map<String, String> serviceRegistryParam) {
    this.serviceRegistryClass = serviceRegistryClass;
    this.serviceRegistryParam = serviceRegistryParam;
  }

  public static XxlRpcInvokerFactory getInstance() {
    return instance;
  }

  // ---------------------- service registry ----------------------

  public void start() throws Exception {
    // start registry
    if (serviceRegistryClass != null) {
      serviceRegistry = serviceRegistryClass.newInstance();
      serviceRegistry.start(serviceRegistryParam);
    }
  }

  public void stop() throws Exception {
    // stop registry
    if (serviceRegistry != null) {
      serviceRegistry.stop();
    }

    // stop callback
    if (CollUtil.isNotEmpty(stopCallbackList)) {
      for (BaseCallback callback : stopCallbackList) {
        try {
          callback.run();
        } catch (Exception e) {
          logger.error(e.getMessage(), e);
        }
      }
    }

    // stop CallbackThreadPool
    stopCallbackThreadPool();
  }

  // ---------------------- future-response pool ----------------------

  // XxlRpcFutureResponseFactory

  public ServiceRegistry getServiceRegistry() {
    return serviceRegistry;
  }

  public void addStopCallBack(BaseCallback callback) {
    stopCallbackList.add(callback);
  }

  public void setInvokerFuture(String requestId, XxlRpcFutureResponse futureResponse) {
    futureResponsePool.put(requestId, futureResponse);
  }

  public void removeInvokerFuture(String requestId) {
    futureResponsePool.remove(requestId);
  }

  // ---------------------- response callback ThreadPool ----------------------

  public void notifyInvokerFuture(String requestId, final XxlRpcResponse xxlRpcResponse) {

    // get
    final XxlRpcFutureResponse futureResponse = futureResponsePool.get(requestId);
    if (futureResponse == null) {
      return;
    }

    // notify
    if (futureResponse.getInvokeCallback() != null) {

      // callback type
      try {
        executeResponseCallback(
            () -> {
              if (xxlRpcResponse.getErrorMsg() != null) {
                futureResponse
                    .getInvokeCallback()
                    .onFailure(new XxlRpcException(xxlRpcResponse.getErrorMsg()));
              } else {
                futureResponse.getInvokeCallback().onSuccess(xxlRpcResponse.getResult());
              }
            });
      } catch (Exception e) {
        logger.error(e.getMessage(), e);
      }
    } else {

      // other nomal type
      futureResponse.setResponse(xxlRpcResponse);
    }

    // do remove
    futureResponsePool.remove(requestId);
  }

  public void executeResponseCallback(Runnable runnable) {
    if (responseCallbackThreadPool == null) {
      synchronized (this) {
        if (responseCallbackThreadPool == null) {
          responseCallbackThreadPool =
              new ThreadPoolExecutor(
                  10,
                  100,
                  60L,
                  TimeUnit.SECONDS,
                  new LinkedBlockingQueue<>(1000),
                  r ->
                      new Thread(
                          r,
                          "xxl-rpc, XxlRpcInvokerFactory-responseCallbackThreadPool-"
                              + r.hashCode()),
                  (r, executor) -> {
                    throw new XxlRpcException("xxl-rpc Invoke Callback Thread pool is EXHAUSTED!");
                  }); // default maxThreads 300, minThreads 60
        }
      }
    }
    responseCallbackThreadPool.execute(runnable);
  }

  public void stopCallbackThreadPool() {
    if (responseCallbackThreadPool != null) {
      responseCallbackThreadPool.shutdown();
    }
  }
}
